package main

// func main() {
// 	// 连接到RabbitMQ服务器
// 	conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
// 	if err != nil {
// 		log.Fatalf("无法连接到RabbitMQ服务器：%s", err)
// 	}
// 	defer conn.Close()

// 	// 创建一个通道
// 	ch, err := conn.Channel()
// 	if err != nil {
// 		log.Fatalf("无法创建通道：%s", err)
// 	}
// 	defer ch.Close()
// 	// 启动多个消费者并行处理任务
// 	for i := 1; i <= 3; i++ {
// 		go startConsumer(i, ch)
// 	}
// 	// 阻塞主进程
// 	select {}
// }

// func generateTask(id int) string {
// 	time.Sleep(time.Duration(rand.Intn(3)) * time.Second)
// 	return fmt.Sprintf("Task %d", id)
// }

// func startConsumer(id int, ch *amqp.Channel) {
// 	// 声明一个队列
// 	queue, err := ch.QueueDeclare(
// 		"tasks_queue", // 队列名
// 		true,          // 持久性
// 		false,         // 自动删除
// 		false,         // 独占
// 		false,         // 等待服务器确认
// 		nil,           // 参数
// 	)
// 	if err != nil {
// 		log.Fatalf("无法声明队列：%s", err)
// 	}
// 	// 消费任务
// 	msgs, err := ch.Consume(
// 		queue.Name, // 队列名
// 		"",         // 消费者标签
// 		false,      // 手动确认
// 		false,      // 不等待服务器确认
// 		false,      // 不使用内置的参数
// 		false,      // 参数
// 		nil,           // 参数
// 	)
// 	if err != nil {
// 		log.Fatalf("无法注册消费者：%s", err)
// 	}
// 	for msg := range msgs {
// 		task := string(msg.Body)
// 		log.Printf("消费者 %d 接收到任务：%s", id, task)
// 		log.Printf("消费者 %d 完成任务：%s", id, task)
// 		// 手动确认任务已处理
// 		msg.Ack(false)
// 	}
// }

///////////////////// 订阅模式消费者

// func failOnError(err error, msg string) {
// 	if err != nil {
// 		log.Fatalf("%s: %s", msg, err)
// 	}
// }

// func main() {
// 	// 连接到RabbitMQ服务器
// 	conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
// 	failOnError(err, "Failed to connect to RabbitMQ")
// 	defer conn.Close()

// 	// 创建一个通道
// 	ch, err := conn.Channel()
// 	failOnError(err, "Failed to open a channel")
// 	defer ch.Close()

// 	// 声明一个交换机
// 	err = ch.ExchangeDeclare(
// 		"logs",   // 交换机名称
// 		"fanout", // 交换机类型
// 		true,     // 是否持久化
// 		false,    // 是否自动删除
// 		false,    // 是否内部使用
// 		false,    // 是否等待服务器响应
// 		nil,      // 其他属性
// 	)
// 	failOnError(err, "Failed to declare an exchange")

// 	// 声明一个临时队列
// 	q, err := ch.QueueDeclare(
// 		"",    // 队列名称，留空表示由RabbitMQ自动生成
// 		false, // 是否持久化
// 		false, // 是否自动删除（当没有任何消费者连接时）
// 		true,  // 是否排他队列（仅限于当前连接）
// 		false, // 是否等待服务器响应
// 		nil,   // 其他属性
// 	)
// 	failOnError(err, "Failed to declare a queue")

// 	// 将队列绑定到交换机上
// 	err = ch.QueueBind(
// 		q.Name, // 队列名称
// 		"",     // 路由键，留空表示接收交换机的所有消息
// 		"logs", // 交换机名称
// 		false,  // 是否等待服务器响应
// 		nil,    // 其他属性
// 	)
// 	failOnError(err, "Failed to bind a queue")

// 	// 订阅消息
// 	msgs, err := ch.Consume(
// 		q.Name, // 队列名称
// 		"",     // 消费者标识符，留空表示由RabbitMQ自动生成
// 		true,   // 是否自动应答
// 		false,  // 是否独占模式（仅限于当前连接）
// 		false,  // 是否等待服务器响应
// 		false,  // 其他属性
// 		nil,    // 其他属性
// 	)
// 	failOnError(err, "Failed to register a consumer")

// 	// 接收消息的goroutine
// 	go func() {
// 		for d := range msgs {
// 			log.Printf("Received a message: %s", d.Body)
// 		}
// 	}()

// 	log.Printf("Waiting for messages. To exit press CTRL+C")
// 	<-make(chan struct{}) // 阻塞主goroutine
// }

// //////////////////routing model
// func failOnError(err error, msg string) {
// 	if err != nil {
// 		log.Fatalf("%s: %s", msg, err)
// 	}
// }

// func main() {
// 	// 连接到RabbitMQ服务器
// 	conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
// 	failOnError(err, "Failed to connect to RabbitMQ")
// 	defer conn.Close()

// 	// 创建一个通道
// 	ch, err := conn.Channel()
// 	failOnError(err, "Failed to open a channel")
// 	defer ch.Close()

// 	// 声明一个交换机
// 	err = ch.ExchangeDeclare(
// 		"logs_direct", // 交换机名称
// 		"direct",      // 交换机类型
// 		true,          // 是否持久化
// 		false,         // 是否自动删除
// 		false,         // 是否内部使用
// 		false,         // 是否等待服务器响应
// 		nil,           // 其他属性
// 	)
// 	failOnError(err, "Failed to declare an exchange")

// 	// 声明一个临时队列
// 	q, err := ch.QueueDeclare(
// 		"",    // 队列名称，留空表示由RabbitMQ自动生成
// 		false, // 是否持久化
// 		false, // 是否自动删除（当没有任何消费者连接时）
// 		true,  // 是否排他队列（仅限于当前连接）
// 		false, // 是否等待服务器响应
// 		nil,   // 其他属性
// 	)
// 	failOnError(err, "Failed to declare a queue")

// 	// 从命令行参数获取要绑定的路由键
// 	if len(os.Args) < 2 {
// 		log.Fatalf("Usage: %s [info] [warning] [error]", os.Args[0])
// 	}
// 	severities := os.Args[1:]

// 	// 将队列绑定到交换机上，并指定要接收的路由键
// 	for _, severity := range severities {
// 		err = ch.QueueBind(
// 			q.Name,        // 队列名称
// 			severity,      // 路由键
// 			"logs_direct", // 交换机名称
// 			false,         // 是否等待服务器响应
// 			nil,           // 其他属性
// 		)
// 		failOnError(err, "Failed to bind a queue")
// 	}

// 	// 订阅消息
// 	msgs, err := ch.Consume(
// 		q.Name, // 队列名称
// 		"",     // 消费者标识符，留空表示由RabbitMQ自动生成
// 		true,   // 是否自动应答
// 		false,  // 是否独占模式（仅限于当前连接）
// 		false,  // 是否等待服务器响应
// 		false,  // 其他属性
// 		nil,    // 其他属性
// 	)
// 	failOnError(err, "Failed to register a consumer")

// 	// 接收消息的goroutine
// 	go func() {
// 		for d := range msgs {
// 			log.Printf("Received a message: %s", d.Body)
// 		}
// 	}()

// 	log.Printf("Waiting for messages. To exit press CTRL+C")
// 	<-make(chan struct{}) // 阻塞主goroutine
// }

// ////////////////////topic

// func failOnError(err error, msg string) {
// 	if err != nil {
// 		log.Fatalf("%s: %s", msg, err)
// 	}
// }

// func main() {
// 	// 连接到RabbitMQ服务器
// 	conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
// 	failOnError(err, "Failed to connect to RabbitMQ")
// 	defer conn.Close()

// 	// 创建一个通道
// 	ch, err := conn.Channel()
// 	failOnError(err, "Failed to open a channel")
// 	defer ch.Close()

// 	// 声明一个交换机
// 	err = ch.ExchangeDeclare(
// 		"logs_topic", // 交换机名称
// 		"topic",      // 交换机类型
// 		true,         // 是否持久化
// 		false,        // 是否自动删除
// 		false,        // 是否内部使用
// 		false,        // 是否等待服务器响应
// 		nil,          // 其他属性
// 	)
// 	failOnError(err, "Failed to declare an exchange")

// 	// 声明一个临时队列
// 	q, err := ch.QueueDeclare(
// 		"",    // 队列名称，留空表示由RabbitMQ自动生成
// 		false, // 是否持久化
// 		false, // 是否自动删除（当没有任何消费者连接时）
// 		true,  // 是否排他队列（仅限于当前连接）
// 		false, // 是否等待服务器响应
// 		nil,   // 其他属性
// 	)
// 	failOnError(err, "Failed to declare a queue")

// 	// 每次取一个 的设置
// 		ch.Qos(
// 			1,
// 			0,
// 			false,
// 		)
// 	// 将队列绑定到交换机上，并指定要接收的路由键
// 	err = ch.QueueBind(
// 		q.Name,       // 队列名称
// 		"example.#",  // 路由键，可以使用通配符*匹配多个单词
// 		"logs_topic", // 交换机名称
// 		false,        // 是否等待服务器响应
// 		nil,          // 其他属性
// 	)
// 	failOnError(err, "Failed to bind a queue")

// 	// 创建一个消费者通道
// 	msgs, err := ch.Consume(
// 		q.Name, // 队列名称
// 		"",     // 消费者标识符，留空表示由RabbitMQ自动生成
// 		true,   // 是否自动应答
// 		false,  // 是否排他消费者
// 		false,  // 是否阻塞
// 		false,  // 是否等待服务器响应
// 		nil,    // 其他属性
// 	)
// 	failOnError(err, "Failed to register a consumer")

// 	// 接收和处理消息
// 	forever := make(chan bool)

// 	go func() {
// 		for d := range msgs {
// 			log.Printf("Received a message: %s", d.Body)
// 		}
// 	}()

// 	log.Printf("Waiting for messages...")
// 	<-forever
// }

// /、////////y延时队列

import (
	"log"

	"github.com/streadway/amqp"
)

func failOnError(err error, msg string) {
	if err != nil {
		log.Fatalf("%s: %s", msg, err)
	}
}


func main() {
	// 建立链接
	conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
	failOnError(err, "Failed to connect to RabbitMQ")
	defer conn.Close()

	ch, err := conn.Channel()
	failOnError(err, "Failed to open a channel")
	defer ch.Close()

	// 声明一个主要使用的 exchange
	err = ch.ExchangeDeclare(
		"logs",   // name
		"fanout", // type
		true,     // durable
		false,    // auto-deleted
		false,    // internal
		false,    // no-wait
		nil,      // arguments
	)
	failOnError(err, "Failed to declare an exchange")

	// 声明一个常规的队列, 其实这个也没必要声明,因为 exchange 会默认绑定一个队列
	q, err := ch.QueueDeclare(
		"test_logs",    // name
		false, // durable
		false, // delete when unused
		true,  // exclusive
		false, // no-wait
		nil,   // arguments
	)
	failOnError(err, "Failed to declare a queue")

    /**
     * 注意,这里是重点!!!!!
     * 声明一个延时队列, ß我们的延时消息就是要发送到这里
     */
	_, errDelay := ch.QueueDeclare(
		"test_delay",    // name
		false, // durable
		false, // delete when unused
		true,  // exclusive
		false, // no-wait
		amqp.Table{
			// 当消息过期时把消息发送到 logs 这个 exchange
			"x-dead-letter-exchange":"logs",
		},   // arguments
	)
	failOnError(errDelay, "Failed to declare a delay_queue")

	err = ch.QueueBind(
		q.Name, // queue name, 这里指的是 test_logs
		"",     // routing key
		"logs", // exchange
		false,
		nil)
	failOnError(err, "Failed to bind a queue")

	// 这里监听的是 test_logs
	msgs, err := ch.Consume(
		q.Name, // queue name, 这里指的是 test_logs
		"",     // consumer
		true,   // auto-ack
		false,  // exclusive
		false,  // no-local
		false,  // no-wait
		nil,    // args
	)
	failOnError(err, "Failed to register a consumer")

	forever := make(chan bool)

	go func() {
		for d := range msgs {
			log.Printf(" [x] %s", d.Body)
		}
	}()

	log.Printf(" [*] Waiting for logs. To exit press CTRL+C")
	<-forever
}
